package com.gzx.asr.modular.funasr.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.gzx.asr.client.FunAsrWsClient;
import com.gzx.asr.constant.ModeConstant;
import com.gzx.asr.modular.funasr.service.FunAsrJobService;
import com.gzx.asr.modular.media.service.GzxHtCallReceiptBusinessService;
import com.gzx.asr.modular.media.vo.GzxHtCallReceiptBusinessVo;
import com.gzx.asr.modular.text.service.GzxAsrVoiceTextService;
import com.gzx.asr.modular.text.vo.GzxAsrVoiceTextVo;
import com.gzx.asr.utils.AudioSplitterUtil;
import com.gzx.asr.utils.DownloadFileUtil;
import com.gzx.asr.utils.FileUtils;
import com.gzx.asr.utils.TimestampUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.io.File;
import java.net.URI;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author xbl
 */
@Slf4j
@Service
public class FunAsrJobServiceImpl implements FunAsrJobService {

    //话单
    private final GzxHtCallReceiptBusinessService businessService;

    //已处理的话单转文本
    private final GzxAsrVoiceTextService asrVoiceTextService;

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public FunAsrJobServiceImpl(GzxHtCallReceiptBusinessService businessService, GzxAsrVoiceTextService asrVoiceTextService, KafkaTemplate<String, String> kafkaTemplate) {
        this.businessService = businessService;
        this.asrVoiceTextService = asrVoiceTextService;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Value("${parameters.serverIpPort}")
    private String serverIpPort;
    @Value("${parameters.mode}")
    private String mode;

    /**
     * 语音转文字：每次按照最大id扫描出100条
     */
    @Override
    public List<GzxAsrVoiceTextVo> voiceConvertText(List<GzxHtCallReceiptBusinessVo> list) {
        List<GzxAsrVoiceTextVo> textList = new ArrayList<>();
        for (GzxHtCallReceiptBusinessVo businessVo : list) {
            try {
                //本地全路径地址
                String wavPath = businessVo.getFileUrl();
                //判断文件是否存在：音频没有内容的无法切分为两个声道的音频文件
                File file = new File(wavPath);
                if (!file.exists()) {
                    log.error("文件不存在：{}", wavPath);
                }
                //文件名，热词待扩展
                FunAsrWsClient client = new FunAsrWsClient(new URI(serverIpPort), mode, wavPath, businessVo.getCallId(), "");
                client.connect();
                //单个语音文件的所有对话
                List<GzxAsrVoiceTextVo> results = client.getResults();

                for (GzxAsrVoiceTextVo result : results) {
                    GzxAsrVoiceTextVo textVo = new GzxAsrVoiceTextVo();
                    textVo.setHtVoiceId(businessVo.getId());
                    textVo.setCallId(businessVo.getCallId());
                    textVo.setConnectTime(businessVo.getConnectTime());
                    textVo.setChatTime(result.getChatTime());
                    textVo.setContent(result.getContent());
                    textList.add(textVo);
                }
            } catch (Exception e) {
                log.error("语音转文字异常： ", e);
            }
        }
        return textList;
    }

    //根据timestamp的时间差计算spk说话人
    private List<GzxAsrVoiceTextVo> splitSpeak(List<GzxAsrVoiceTextVo> textList) {
        //根据callId分组之后处理
        Map<String, List<GzxAsrVoiceTextVo>> callIdMap = textList.stream()
                .collect(Collectors.groupingBy(GzxAsrVoiceTextVo::getCallId));
        for (Map.Entry<String, List<GzxAsrVoiceTextVo>> entry : callIdMap.entrySet()) {
            List<GzxAsrVoiceTextVo> list = entry.getValue();
            //计算说话人
            evalSpeaker(list);
        }
        //设置spk-role
//        setSpkTag(textList);

        return textList;
    }

    //计算说话人
    private static void evalSpeaker(List<GzxAsrVoiceTextVo> textList) {
        //上一次说话人是谁，默认是客服
        int lastSpk = 0;
        //根据timestamp的时间差计算spk说话人
        for (int i = 0; i < textList.size() - 1; i += 2) {
            GzxAsrVoiceTextVo current = textList.get(i);
            GzxAsrVoiceTextVo next = textList.get(i + 1);

            // 判断是否属于同一个说话人
            boolean isSameSpeaker = TimestampUtil.isSameSpeaker(current.getChatTime(), next.getChatTime());

            if (isSameSpeaker) {
                // 如果是同一人，都设为客服（或根据业务逻辑决定）
                current.setSpeakerRole(1); // 1 是客服
                next.setSpeakerRole(1);
            } else {
                // 否则交替设置不同角色，比如客服、客户轮流
                current.setSpeakerRole(1); // 客服
                next.setSpeakerRole(2);     // 客户
            }
        }
        // 处理最后一个未被处理的元素（如果总数为奇数）
        if (textList.size() % 2 != 0) {
            GzxAsrVoiceTextVo last = textList.getLast();
            last.setSpeakerRole(1); // 或者根据上一个判断来设定
        }

        //把时间戳转化为当前说话时间(年月日时分秒毫秒)
        for (GzxAsrVoiceTextVo textVo : textList) {
            if (!ObjectUtils.isEmpty(textVo.getChatTime())) {
                textVo.setChatTime(TimestampUtil.getTalkTimeStamp(textVo.getConnectTime(), TimestampUtil.getTimeStamp(textVo.getChatTime())));
            }
        }
    }

    //设置spk-role标签，修改关键字
    private static void setSpkTag(List<GzxAsrVoiceTextVo> textList) {
        //说话内容含有“调解中心”是客服，客服的spk-role设置为1，客户设置为2
        List<String> spkList = textList.stream()
                .filter(item -> item.getContent().contains(ModeConstant.SPEAK_KEYWORD)) // 过滤 content 包含 "调解中心" 的对象
                .map(GzxAsrVoiceTextVo::getSpk) // 提取 spk
                .distinct() // 去重（如果需要）
                .toList();
        //说话内容含有“调解中心”的说话人
        String speaker;
        if (!spkList.isEmpty()) {
            speaker = spkList.getFirst();
        } else {
            speaker = "";
        }
        log.info("说话人{}是客服", speaker);
        textList.forEach(item -> {
            //说话内容含有“调解中心”的说话人是客服
            if (ModeConstant.SPEAK_ONE.equals(speaker)) {
                item.setSpeakerRole(1);
            } else if (ModeConstant.SPEAK_TWO.equals(speaker)) {
                item.setSpeakerRole(2);
            } else {
                //未接通等情况默认是客服
                item.setSpeakerRole(1);
            }
        });
    }

    @Override
    public void voiceToTextJob() {
        //每次获取上次处理之后的最大id
        int maxVoiceId = asrVoiceTextService.getMaxVoiceId();
        //每次按照最大id扫描出100条
        List<GzxHtCallReceiptBusinessVo> list = businessService.getVoiceByMaxId(maxVoiceId);
        if (list.isEmpty()) {
            return;
        }
        //下载的wav文件地址
        List<GzxHtCallReceiptBusinessVo> downList = downloadWav(list);

        //调用asr转换文本
        List<GzxAsrVoiceTextVo> textList = voiceConvertText(downList);
        if (textList.isEmpty()) {
            return;
        }
        //根据timestamp的时间差计算spk说话人
        List<GzxAsrVoiceTextVo> speakList = splitSpeak(textList);

        //按照callId和chatTime排序
        speakList = speakList.stream()
                //  过滤掉空的ChatTime
                .filter(item -> !ObjectUtils.isEmpty(item.getChatTime()))
                .sorted(Comparator.comparing(GzxAsrVoiceTextVo::getCallId)
                        .thenComparing(GzxAsrVoiceTextVo::getChatTime)).toList();
        //拿到文本存库
        asrVoiceTextService.batchInsert(speakList);
    }

    //把录音文件转化为文本发送到Kafka，质检消费之后入库
    @Override
    public void voiceToText(List<GzxHtCallReceiptBusinessVo> list) {
        Thread.startVirtualThread(() -> {
            if (CollUtil.isNotEmpty(list)) {
                //下载的wav文件地址
                List<GzxHtCallReceiptBusinessVo> downList = downloadWav(list);

                //调用asr转换文本
                List<GzxAsrVoiceTextVo> textList = voiceConvertText(downList);
                if (textList.isEmpty()) {
                    return;
                }
                //根据timestamp的时间差计算spk说话人
                List<GzxAsrVoiceTextVo> speakList = splitSpeak(textList);

                //按照callId和chatTime排序
                speakList = speakList.stream()
                        //  过滤掉空的ChatTime
                        .filter(item -> !ObjectUtils.isEmpty(item.getChatTime()))
                        .sorted(Comparator.comparing(GzxAsrVoiceTextVo::getCallId)
                                .thenComparing(GzxAsrVoiceTextVo::getChatTime)).toList();
                if (!speakList.isEmpty()) {
                    //发到kafka，一次发list数据太多，会造成数据截断，需要遍历list发送
                    log.info("发到kafka的asr-text-topic {} 条", speakList.size());
                    for (GzxAsrVoiceTextVo item : speakList) {
                        //没有说话内容的丢弃
                        if (!ObjectUtils.isEmpty(item)) {
                            kafkaTemplate.send("asr-text-topic", "asr-text", JSON.toJSONString(item));
                        }
                    }
                }
            }
        });
    }

    //下载文件到本地
    private List<GzxHtCallReceiptBusinessVo> downloadWav(List<GzxHtCallReceiptBusinessVo> list) {
        List<GzxHtCallReceiptBusinessVo> resultList = new ArrayList<>();
        try {
            for (GzxHtCallReceiptBusinessVo businessVo : list) {
                String url = DownloadFileUtil.downloadFile(businessVo.getFileUrl());

                //封装vo
                GzxHtCallReceiptBusinessVo vo = new GzxHtCallReceiptBusinessVo();
                vo.setId(businessVo.getId());
                vo.setCallId(businessVo.getCallId());
                vo.setConnectTime(businessVo.getConnectTime());
                vo.setFileUrl(url);
                resultList.add(vo);
            }
        } catch (Exception e) {
            log.error("下载文件到本地异常 ", e);
        }
        return resultList;
    }

    //音频分声道
    private List<GzxHtCallReceiptBusinessVo> splitTwoChannels(List<GzxHtCallReceiptBusinessVo> list) {
        List<GzxHtCallReceiptBusinessVo> resultList = new ArrayList<>();
        try {
            for (GzxHtCallReceiptBusinessVo businessVo : list) {
                String url = DownloadFileUtil.downloadFile(businessVo.getFileUrl());
                AudioSplitterUtil.splitTwoVoice(url);

                // 创建左声道副本
                GzxHtCallReceiptBusinessVo leftVo = new GzxHtCallReceiptBusinessVo();
                leftVo.setId(businessVo.getId());
                leftVo.setCallId(businessVo.getCallId());
                leftVo.setConnectTime(businessVo.getConnectTime());
                leftVo.setFileUrl(FileUtils.splitLeftChannel(url));
                resultList.add(leftVo);

                // 创建右声道副本
                GzxHtCallReceiptBusinessVo rightVo = new GzxHtCallReceiptBusinessVo();
                rightVo.setId(businessVo.getId());
                rightVo.setCallId(businessVo.getCallId());
                rightVo.setConnectTime(businessVo.getConnectTime());
                rightVo.setFileUrl(FileUtils.splitRightChannel(url));
                resultList.add(rightVo);
            }
        } catch (Exception e) {
            log.error("音频拆分左右声道异常 ", e);
        }
        return resultList;
    }

}
